package org.raymond.iworks.study.basic.thread.flow;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

/**
 * 1. Publisher 发布者，进行数据发布
 * 1.1 subscribe 发布者与订阅者建立订阅关系
 * 2. Subscriber 订阅者，订阅数据
 * 2.1 onSubscribe 发布者通知订阅者开始发送数据
 * 2.2 onNext 订阅者接收数据
 * 2.3 onError 出现异常
 * 2.4 onComplete 数据处理完成
 * 3. Subscription 数据订阅，请求数据、取消订阅
 * 3.1 request
 * 3.2 cancel
 * 4. Processor 数据的中间操作，既是Subscriber也是Publisher
 */
@Slf4j
public class Demo1 {
    public static void main(String[] args) {
        test1();
    }
    // 订阅/发布
    public static void test1(){
        //数据发布者
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        //中间处理者
        FlowProcessor processor = new FlowProcessor();
        //数据订阅者
        Flow.Subscriber<String> subscriber = new FlowSubscriber();

        //发布者与中间处理者建立关系
        publisher.subscribe(processor);
        //中间处理者与订阅者建立关系
        processor.subscribe(subscriber);

        //发布者开始发布数据
        for (int i = 0; i < 10; i++) {
            String msg = "hello flow: " + i;
            log.info("发布者发送数据" + i);
            publisher.submit(msg);
        }

        //关闭发布者
        publisher.close();

        //休眠等待，防止主线程退出
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
